package com.xctech.yace.manger;

import com.xctech.yace.anno.DataMapping;
import com.xctech.yace.pojo.GlobalClass;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.ibatis.session.SqlSession;

import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;


/**
 * 数据管理类基类
 *
 * @author junfeng.chen@xuncetech.com
 * @date 2021/10/27
 */
@Data
@Slf4j
public abstract class DataManager {

    /**
     * 线城池、sharedLock以及methodsMap的容量
     */
    private static final int MAX_SIZE = 5;

    /**
     * key：表名，value：HashMap<String,Object> 主键 ：单条数据
     */
    public volatile Map<String, HashMap<String, Object>> datasCache = new HashMap<>();

    /**
     * 接口方法集合
     *  key = 表名.hashCode() % MAX_SIZE
     */
    public volatile Map<Integer, List<Method>> methodsMap = new HashMap<>(MAX_SIZE);

    public volatile Map<String, Map<String, List<String>>> Map = new HashMap<>(MAX_SIZE);

    /**
     * 方法-主鍵映射集合
     */
    public volatile Map<String, String> methodsKeysMap = new HashMap<>();

    /**
     * 互斥锁
     */
    public final ReentrantLock lock = new ReentrantLock();


    /**
     * 加載數據
     * 需要各实现类实现
     * @param sqlSession
     */
    public abstract void loadData(SqlSession sqlSession);

    /**
     * 初始化
     *
     * @param mapperClass    DataMapping接口
     * @param parametersMaps DataMapping参数集合
     * @param isInitModel 是否日初模式,日初如果只配置1张表, 只运行该方法,其他方法运行会报错
     *                       例：
     *                       Map<String,Map<String,Object>>   parametersMaps =   new HashMap<String,Map<String,Object>>();
     *                       Map<String,Object> selectVarScrInfosParameters = new HashMap<String,Object>();
     *                       selectVarScrInfosParameters.put("scr_cd",112333);
     *                       parametersMaps.put("selectVarScrInfos",selectVarScrInfosParameters);
     * @DataMapping(element = DimMapping.VAR_SCR_INFO)
     * List<Object> selectVarScrInfos(@Param("map")HashMap map);
     */
    protected void initData(Object mapperClass, Map<String, Map<String, Object>> parametersMaps, boolean isInitModel) {
        /**
         * Flink应用在物理执行图Task部署、提交阶段，调用某个算子的operator.open()，
         * 控制每个TM上多并行时，只有一个并行执行open方法的代码（如果不控制默认按照并行度的个数执行open方法的代码多次）
         *
         * 源码：Task.run()--->invokable.invoke()--->StreamTask.invoke()--->beforeInvoke()---->
         *  operatorChain.initializeStateAndOpenOperators(createStreamTaskStateInitializer());---->operator.open();
         */
        lock.lock();

        try {
            if (methodsMap.isEmpty()) {
                Class<?> anInterface = mapperClass.getClass().getInterfaces()[0];
                //System.out.println(anInterface);
                /**
                 * 动态把dimMapper的所有方法归类放在集合中，类型（key）相同的在同一个线程中提交
                 * key = 表名.hashCode() % MAX_SIZE， 得到一个1-MAX_SIZE的值
                 */
                for (Method method : anInterface.getMethods()) {
                    if (method.isAnnotationPresent(DataMapping.class)) {
                        String table = method.getAnnotation(DataMapping.class).element().getTable();
                        if (isInitModel && !GlobalClass.initModelCalcutables.contains(table)) continue;//日初只运行配置table的方法
                        Integer key = Math.abs(table.hashCode() % MAX_SIZE);
                        if (methodsMap.containsKey(key)) {
                            methodsMap.get(key).add(method);
                        } else {
                            List<Method> methods = new ArrayList();
                            methods.add(method);
                            methodsMap.put(key, methods);
                        }
                    }
                }
            } else {
                // 不为空，直接返回，释放锁
                return;
            }
        } finally {
            lock.unlock();
        }

        log.error("methodsMap size : {}", methodsMap.size());

        /**
         *异步提交 有返回值
         */
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            /**
             * 把动态集合多线程提交，异步加载数据
             */
            ExecutorService es = new ThreadPoolExecutor(MAX_SIZE, MAX_SIZE, 0L, TimeUnit.MILLISECONDS,
                    new ArrayBlockingQueue<>(1), new ThreadPoolExecutor.AbortPolicy());
            /**
             * 共享锁  sharedLock的count和MAX_SIZE一致
             */
            CountDownLatch sharedLock = new CountDownLatch(methodsMap.size());
            try {
                for (Map.Entry<Integer, List<Method>> e : methodsMap.entrySet()) {
                   es.submit(() -> {
                        try {
                            for (Method method : e.getValue()) {
                                String table = method.getAnnotation(DataMapping.class).element().getTable();
                                String[] keys = method.getAnnotation(DataMapping.class).element().getKey();
                                List<Object> invoke;
                                if (parametersMaps == null) {
                                    invoke = (List) method.invoke(mapperClass);
                                } else {
                                    if (parametersMaps.containsKey(method.getName())) {
                                        invoke = (List) method.invoke(mapperClass, parametersMaps.get(method.getName()));
                                    } else {
                                        invoke = (List) method.invoke(mapperClass);
                                    }
                                }
                                log.warn("read table : {}, size : {}", table, invoke.size());
                                //根据主键转换集合中的存储结果
                                transform(invoke, datasCache, table, keys);
                            }
                        } catch (Exception ex) {
                            log.error(ExceptionUtils.getStackTrace(ex));
                        } finally {
                            /**
                             * sharedLock的count--
                             */
                            sharedLock.countDown();
                            log.warn("Decrements the count of the latch,the current count is :{}", sharedLock.getCount());
                        }
                    });
                }
            } catch (Exception e) {
                log.error(ExceptionUtils.getStackTrace(e));
            } finally {
                try {
                    /**
                     * 阻塞，等待count=0時喚醒
                     */
                    sharedLock.await();
                    log.warn("Decrements the count of the latch,the current count is :{}", sharedLock.getCount());
                } catch (Exception e) {
                    log.error(ExceptionUtils.getStackTrace(e));
                }
                es.shutdown();
            }
            return "over!";
        });

        try {
            /**
             * 等待返回结果
             */
            String result = future.get();
            if(isInitModel)
                log.warn("Fact initData is :{}", result);
            else
                log.warn("Dim initData is :{}", result);
        }catch (Exception ex){
            log.error(ExceptionUtils.getStackTrace(ex));
        }
        if(isInitModel)
          log.warn("Fact initData is over over ！！！");
        else
          log.warn("Dim initData is over over ！！！");
    }

    /**
     * 根据主键转换数据
     */
    private static void transform(List<Object> objects, Map<String, HashMap<String, Object>> datasCache, String name, String[] keys) {
        HashMap<String, Object> val = new HashMap<>();

        for (Object object : objects) {
            for (String key : keys) {
                String joinKey = String.valueOf(System.nanoTime());
                StringBuilder strBuilder = new StringBuilder();
                // 配置多个唯一索引值，循环存入
                if (StringUtils.isNotBlank(key)) {
                    for (String k : key.split("\\|")) {
                        Class<?> aClass = object.getClass();
                        Field declaredField;
                        try {
                            declaredField = aClass.getDeclaredField(k);
                            declaredField.setAccessible(true);
                            Object o = declaredField.get(object);
                            strBuilder.append(o).append("_");
                        } catch (Exception e) {
                            log.error("反射转换表:{}，主键key:{}数据异常：{}",name,key,ExceptionUtils.getStackTrace(e));
                        }
                    }
                    joinKey = strBuilder.substring(0, strBuilder.length() - 1);
                }
                val.put(joinKey,object);
            }
            datasCache.put(name, val);
        }

    }

    /**
     * 根据表名获取表记录
     *
     * @param name
     * @return
     */
    public HashMap<String, Object> getTableData(String name) {
        return datasCache.getOrDefault(name, new HashMap<>());
    }

    /**
     * 根据表名，主键获取指定记录
     *了；
     * @param name
     * @param key
     * @return
     */
    public <E> E getRecord(String name, String key) {
        HashMap<String, Object> map = datasCache.get(name);
        if(map != null){
            Object o = map.get(key);
            if(o != null){
                return (E)o;
            }
        }
        return null;
    }

    public <E> E getRecordAndCastType(String name, String key) {
        HashMap<String, Object> stringObjectHashMap = datasCache.get(name);
        if(stringObjectHashMap==null) return null;
        Object o = stringObjectHashMap.get(key);
        if (o==null)
            return null;
        return (E) o;
    }
}
